/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.sss.common.SentinelThreadPool;
import org.sss.common.io.IOThread;
import org.sss.common.io.KeyFormatter;
import org.sss.common.io.StreamDB;
import org.sss.mapreduce.DB;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.mapreduce.TTDB;

/**
 * Server side of 'rmtuplegroup.sh' implementation.
 */
public class CommandRemoveTupleGroup {
  private static Logger logger = SssServer.getLogger(CommandRemoveTupleGroup.class);

  public static long execute(SentinelThreadPool iotp,
                             List<StorageNode> storageNodes,
                             final GroupID gid) throws SssException, InterruptedException {
    final AtomicLong count = new AtomicLong(0);
    final CountDownLatch latch = new CountDownLatch(storageNodes.size());
    final List<Throwable> exps = new ArrayList<Throwable>();

    for (final StorageNode node: storageNodes) {
      iotp.execute(new Runnable() {
        @Override
        public void run() {
          try {
            final IOThread iot = IOThread.currentThread();
            if (iot.isStreamDB()) {
              final StreamDB db = iot.getStreamDB(node);
              db.delete(gid);
            }
            else {
              // TODO: makes sssConf.io_out_batchcount?
              final int count_max = iot.getIOGetBatchCount();
              final DB db = iot.getDB(node);
              final byte[] first = KeyFormatter.getGroupIDPrefix(gid);
              long total = 0L;
              if (iot.isRestrictedDBAccess() && db.getClass() == TTDB.class) {
                final List<byte[]> allkeys = db.fwmkeys(first, -1);
                final int count_target = allkeys.size();
                boolean done = count_target == 0;
                int count_sofar = 0;
                while (!done) {
                  final int count_tmp = Math.min(count_target - count_sofar, count_max);
                  byte[][] ks = new byte[count_tmp][];
                  for (int i=0; i<count_tmp; i++) {
                    ks[i] = allkeys.get(count_sofar + i);
                  }
                  db.outlist(ks);
                  count_sofar += count_tmp;
                  done = count_target == count_sofar;
                }
                total = count_target;
              }
              else {
                boolean initial = true;
                long removed = 0L;
                final byte[] sentinel = KeyFormatter.getGroupIDSentinel(gid);
                do {
                  removed = db.rangeout(first, sentinel, -1, initial);
                  initial = false;
                  total += removed;
                }
                while (removed > 0) ;
              }
              count.addAndGet(total);
            }
          } catch (Throwable e) {
              logger.error("", e);
              synchronized (exps) {
                exps.add(e);
              }
          } finally {
            latch.countDown();
          }
        }
      });
    }
    latch.await();

    if (!exps.isEmpty()) {
      throw new SssException(exps.get(0));
    }

    return count.longValue();
  }
}
